package cn.doitedu.flink.demos;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;

import java.time.Duration;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/2/23
 **/
public class WindowApiTest {
    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.addSource(SourcesTest.createSocketSource("doitedu01", 9999));

        SingleOutputStreamOperator<Tuple4<String, String, Integer, Long>> maped = source.map(line -> {
            String[] arr = line.split(",");
            return Tuple4.<String, String, Integer, Long>of(arr[0], arr[1], Integer.parseInt(arr[2]), Long.parseLong(arr[3]));
        }).returns(TypeInformation.of(new TypeHint<Tuple4<String, String, Integer, Long>>() {
        }));

        // 全局计数window
        // maped.countWindowAll(3).sum("f2").print();

        // 自定义count窗口
        // 窗口触发条件为：3条数据 ；  发射出去的窗口：包含4条数据； 意味着这相当于一个滑动窗口：...[3,4,5,6],[6,7,8,9],[9,10,11,12]...  窗口宽度为4，滑动间距为3
        // maped.windowAll(GlobalWindows.create()).trigger(CountTrigger.of(3)).evictor(CountEvictor.of(4)).sum("f2").print(); // 这个方式没搞定，不知怎么用


        // 时间窗口，不指定水位生成器会报错
        // Record has Long.MIN_VALUE timestamp (= no timestamp marker). Is the time characteristic set to 'ProcessingTime', or did you forget to call 'DataStream.assignTimestampsAndWatermarks(...)'?
        // maped.windowAll(TumblingEventTimeWindows.of(Time.milliseconds(2000))).sum("f2").print();


        /**
         *
         * a,a,4,1614073701000
         * a,a,4,1614073702000
         * a,a,4,1614073703000
         * a,a,4,1614073704000
         * a,a,4,1614073705000
         * a,a,4,1614073706000
         * a,a,4,1614073707000
         * a,a,4,1614073708000
         * a,a,4,1614073709000
         * a,a,4,1614073710000
         * a,a,4,1614073711000
         * a,a,4,1614073712000
         * a,a,4,1614073713000
         *
         * 2> (a,a,12,1614073701000)
         * 1> (a,a,12,1614073704000)
         * 2> (a,a,12,1614073707000)
         */
        WatermarkStrategy<Tuple4<String, String, Integer, Long>> strategy = WatermarkStrategy.<Tuple4<String, String, Integer, Long>>forBoundedOutOfOrderness(Duration.ofMillis(2000))
                .withTimestampAssigner((tp, timestamp) -> tp.f3);
        // 事件时间窗口，指定水位生成器
         SingleOutputStreamOperator<Tuple4<String, String, Integer, Long>> watermarks = maped.assignTimestampsAndWatermarks(strategy);
        // watermarks.windowAll(TumblingEventTimeWindows.of(Time.milliseconds(3000))).sum("f2").print();


        /**
         * 数据分区处理
         */
        TypeInformation<Tuple2<String, String>> type = TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
        });
        KeyedStream<Tuple4<String, String, Integer, Long>, Tuple2<String, String>> keyed = maped.keyBy(tp -> Tuple2.<String, String>of(tp.f0, tp.f1), type);

        // 分区后，各key下单独生成窗口来计算
        // keyed.countWindow(3).sum("f2").print();

        // 虽已分区，但按全局来生成窗口并计算
        // keyed.countWindowAll(3).sum("f2").print();

        // 处理时间窗口
        // keyed.window(TumblingProcessingTimeWindows.of(Time.milliseconds(3000))).sum("f2").print();

        // 自定义聚合函数
        AggregateFunction<Tuple4<String, String, Integer, Long>, Integer, Integer> aggregateFunction = new AggregateFunction<Tuple4<String, String, Integer, Long>, Integer, Integer>() {
            @Override
            public Integer createAccumulator() {
                return 0;
            }

            @Override
            public Integer add(Tuple4<String, String, Integer, Long> value, Integer accumulator) {

                return accumulator + value.f2;
            }

            @Override
            public Integer getResult(Integer accumulator) {
                return accumulator;
            }

            @Override
            public Integer merge(Integer a, Integer b) {
                return null;
            }
        };

        // keyed.window(TumblingProcessingTimeWindows.of(Time.milliseconds(3000))).aggregate(aggregateFunction).print();
        KeyedStream<Tuple4<String, String, Integer, Long>, Tuple2<String, String>> keyed2 = watermarks.keyBy(tp -> Tuple2.<String, String>of(tp.f0, tp.f1), type);
        keyed2.window(TumblingEventTimeWindows.of(Time.milliseconds(2000))).reduce(new ReduceFunction<Tuple4<String, String, Integer, Long>>() {
            @Override
            public Tuple4<String, String, Integer, Long> reduce(Tuple4<String, String, Integer, Long> value1, Tuple4<String, String, Integer, Long> value2) throws Exception {

                return Tuple4.of(value2.f0,value2.f1,value1.f2+value2.f2,value2.f3);
            }
        }).print();



        env.execute();
    }
}

